package com.hu.wc

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * @Author: hujianjun
 * @Create Date: 2020/12/2 15:23
 * @Describe: 数据包含三个字段：商品ID,用户ID,访问类型（1.点击查看 2.收藏 3.购买），访问时间；
 *            这里每隔3秒对最近6秒内的数据进行汇总计算各个商品的“点击查看”访问量，也就是访问类型为1的数据
 */

case class ProductLog(productID: String, userID: String, viewType: Int, ts: Long)

case class ProductLogResult(productID: String, windowEnd: String, cnt: Long)

object ProductViewStatistic {
  def main(args: Array[String]): Unit = {
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val inputStream = streamEnv.readTextFile(getClass.getResource("/data/product.csv").getPath)

    val dataStream = inputStream.map(data => {
      val arr = data.split(",")
      ProductLog(arr(0), arr(1), arr(2).toInt, arr(3).toLong)
    }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ProductLog](Time.seconds(1)) {
      override def extractTimestamp(element: ProductLog): Long = element.ts
    })

    val resultStream = dataStream
      //过滤出访问类型为1的记录
      .filter(_.viewType == 1)
      .keyBy(_.productID)
      .timeWindow(Time.seconds(6), Time.seconds(3))
      .aggregate(new ProductViewPreAgg(), new ProductViewResult())

    resultStream.print("统计产品")

    streamEnv.execute("product statistic job")
  }
}

class ProductViewPreAgg extends AggregateFunction[ProductLog, Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(value: ProductLog, accumulator: Long): Long = accumulator + 1

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b
}

class ProductViewResult extends WindowFunction[Long, ProductLogResult, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[ProductLogResult]): Unit = {
    out.collect(ProductLogResult(key, window.getEnd.toString, input.head))
  }
}
